package ds_industry_2025.ds.ds_03.tjxt

import org.apache.spark.mllib.linalg._
import org.apache.spark.mllib.linalg.distributed.RowMatrix
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._

import scala.collection.mutable.ArrayBuffer
/*
      1、根据子任务一的结果，对其进行SVD分解，对数据进行降维保留前5个奇异值信息，根据该用户已购买的商品分别与未购买的商品计算余弦
      相似度再进行累加求均值，将均值最大的5件商品id进行输出作为推荐使用。将输出结果截图粘贴至客户端桌面【Release\任务C提交结果
      .docx】中对应的任务序号下。
结果格式如下：

------------------------推荐Top5结果如下------------------------
相似度top1(商品id：1，平均相似度：0.983456)
相似度top2(商品id：71，平均相似度：0.782672)
相似度top3(商品id：22，平均相似度：0.7635246)
相似度top4(商品id：351，平均相似度：0.7335748)
相似度top5(商品id：14，平均相似度：0.522356)
 */
object t1 {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .master("local[*]")
      .appName("t1")
      .config("hive.exec.dynamic.partition.mode","nonstrict")
      .config("spark.serializer","org.apache.spark.serializer.KryoSerializer")
      .config("spark.sql.extensions","org.apache.spark.sql.hudi.HoodieSparkSessionExtension")
      .enableHiveSupport()
      .getOrCreate()

    import spark.implicits._

    // todo 读取子任务一生成的数据
    val r1 = spark.table("tzgc.skumat")

    //  todo 把用户 0.0的商品购买情况拿到
    val skus:Seq[Double] = r1.filter(col("user_id") === lit(0.0)).limit(1)
      .select(r1.columns.slice(1, r1.columns.length).map(col): _*)
      //  这里需要提取多列，所以要将整行先转化为seq
      .map(r => r.toSeq.map(r => r.toString.toDouble))
      .collect()(0)

    //  todo 创建数组分别存储购买情况和未购买的情况
    var buy=new ArrayBuffer[Double]()
    var no_buy=new ArrayBuffer[Double]()

    skus.zipWithIndex.foreach{
      e => {
        if(e._1.equals(0.0)){
          no_buy += e._2
        }else{
          buy += e._2
        }
      }
    }



    //  todo 将数据转化为Vector，为了方便后面转化为rowMatrix
    val mapData:Array[Vector] = r1.rdd.map(
      r => {
        val v:Vector=Vectors.dense(r.toSeq.slice(1, r.toSeq.length).map(r => r.toString.toDouble).toArray)
        v
      }
    ).collect()

    //  todo 将mapData转化为rowMatrix
    val matrix = new RowMatrix(spark.sparkContext.parallelize(mapData))

    //  todo 对rowMatrix进行svd分解
    val svd = matrix.computeSVD(5, computeU = true)
    //  todo 拿到v(右边奇异值)
    val v=svd.V

    //  todo 将得到的右奇异值向量矩阵，添加索引，转化为数组
    val to_array = v.rowIter.zipWithIndex.toArray

    //  todo 将带索引的奇异值向量矩阵转化为dataframe
    val sku_info = spark.createDataFrame(
      to_array
    ).toDF("vec", "sku_mapping_id")

    sku_info.show

    //  todo 注册计算余弦相似度的函数
    //   todo 由于svd分解之后的右奇异值是一个密集向量，所以这下面和不需要svd分解那里的还是存在去别的
    spark.udf.register(
      "cos",
      (v1:DenseVector,v2:DenseVector) => {
        1 - breeze.linalg.functions.cosineDistance(
            breeze.linalg.DenseVector(v1.values.toArray),
          breeze.linalg.DenseVector(v2.values.toArray)
        )
      }
    )

    //  todo 左边是已经购买的商品 ，右边是未购买的商品
    val result = sku_info.crossJoin(sku_info)
      .toDF("vec", "sku_mapping_id", "vec2", "sku_mapping_id2")
      .filter(col("sku_mapping_id") !== col("sku_mapping_id2"))
      .withColumn("cos", expr("cos(vec,vec2)"))
      .filter(col("sku_mapping_id").isin(buy: _*) && col("sku_mapping_id2").isin((no_buy: _*)))
      .groupBy("sku_mapping_id2")
      .agg(avg("cos").as("avg_cos"))
      .orderBy(desc("avg_cos"))
      .limit(5)

    result.show

    println("-------------------推荐top5结果如下----------------------")
    result.collect().zipWithIndex.foreach{
      case (r,index) =>
        val id=r.getAs[Int]("sku_mapping_id2")
        val cos=r.getAs[Double]("avg_cos")
        val str=s"相似度top${index+1}(商品id:${id},平均相似度:${cos})"
        println(str)
    }


    spark.close()
  }

}
